package com.atguigu.flink.chapter01_wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//导入org.apache.flink.api.common.typeinfo.Types全部的静态内容，例如方法，属性
import java.util.Arrays;

import static org.apache.flink.api.common.typeinfo.Types.*;

//静态导入
// 把某个类的静态方法导入到当前的类中，好比我这个类拥有了导入的静态方法，可以直接使用
/**
 * Created by Smexy on 2022/10/19
 *
 *      如果lamda中丢失的泛型是Tuple之类的，不能使用  returns(Class T)方法，应该用 returns(TypeHint x)
 */
public class Demo6_LamdaTuple2
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //启动一个线程去算。不设置，默认用当前环境中所有的cpu作为线程数进行计算
        //env.setParallelism(1);

        //使用环境读取数据源，获取一个流
        DataStreamSource<String> source = env.socketTextStream("hadoop102", 8888);


        source
            .flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (line, out) -> {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            })
            //.returns(new TypeHint<Tuple2<String, Integer>>() {})
            // Flink提供了一个工具类 Types,保存了常见的 TypeInfomation的实现
            .returns(TUPLE(STRING,INT))
            .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
            .sum(1)
            .print();

        //启动执行环境，计算才会开始，永不结束
        env.execute();

    }
}
